package com.to8to.kitt.esb.client;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.to8to.kitt.client.ThriftClientConfig;
import com.to8to.kitt.esb.BusServer;
import com.to8to.kitt.esb.ZK;
import com.to8to.kitt.utils.DateUtil;
import com.to8to.kitt.utils.LogPatterns;
import com.to8to.kitt.utils.NettyUtils;

public class ESBAddressMonitor {
	
	public static Logger logger = LoggerFactory.getLogger(ESBAddressMonitor.class);
	
	public PathChildrenCache childcache;
	
	public Map<String, ThriftClientConfig> ESBAddresses = new HashMap<String, ThriftClientConfig>();
	
	public ObjectMapper om = new ObjectMapper();
	
	public void start(String path) {
		
		logger.info("监听总线服务: " + path);
		
		childcache = new PathChildrenCache(ZK.ZK_CLIENT, path, true, false, ZK.ZkCuratorThreadPool);
		
		try {
			
			childcache.start();
			
			List<String> instances = ZK.ZK_CLIENT.getChildren().forPath(path);
			
			for(String instance:instances)
			{
				byte[] data = ZK.ZK_CLIENT.getData().forPath(path + "/" + instance);
				
				JsonNode jsonNode = om.readTree(data);
				
            	String host = jsonNode.get("address").asText();
            	
            	int port = jsonNode.get("port").asInt();
            	
            	if(!ESBAddresses.containsKey(instance))
            	{
            		ThriftClientConfig config = new ThriftClientConfig(instance, host, port);
            		ESBAddresses.put(instance, config);
            	}
            	
            	ThriftClientPoolFactory.ensureAdd(instance, host, port);
			}

			childcache.getListenable().addListener(new PathChildrenCacheListener() {

				@Override
				public void childEvent(CuratorFramework client,
						PathChildrenCacheEvent event) throws Exception {
					
					if(event == null)
					{
						logger.error("{} event is null!", LogPatterns.ZK_CONN_TAG);
						return ;
					}
					
					ChildData cdata = event.getData();
					if(cdata == null)
					{
						logger.error("{} cdata is null!", LogPatterns.ZK_CONN_TAG);
						return ;
					}
					
					String path = cdata.getPath();
					
					System.err.println(DateUtil.format(new Date()) + " ZK event path:" + path + ", type:" + event.getType());
					
					
//					ChildData cdata = event.getData();
//					if(cdata == null)
//					{
//						logger.error("cdata is null!");
//						return ;
//					}
//					
//					String path = cdata.getPath();
					
					String instance = StringUtils.substringAfterLast(path, "/");
					
					if(!instance.startsWith("ThriftBusServer"))
						return ;
					
					
					if (event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED) {
						
						byte[] data = cdata.getData();
						
						JsonNode jsonNode = om.readTree(data);
						
		            	String host = jsonNode.get("address").asText();
		            	
		            	int port = jsonNode.get("port").asInt();
		            	
		            	logger.info("add ESB instance: " + instance);
		            	
		            	if(!ESBAddresses.containsKey(instance))
		            	{
		            		ThriftClientConfig config = new ThriftClientConfig(instance, host, port);
		            		ESBAddresses.put(instance, config);
		            	}
		            	
		            	ThriftClientPoolFactory.ensureAdd(instance, host, port);
		            	
					}
					else if (event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
						
						logger.info("remove ESB instance: " + instance);
						
						ESBAddresses.remove(instance);
						
						ThriftClientPoolFactory.remove(instance);
					}
					
				}
			});

		} catch (Throwable e) {
			NettyUtils.exitIfOutOfMemory(e);
			e.printStackTrace();
		}
	}
	
	
	public static void main(String[] args)
	{
		ESBAddressMonitor m = new ESBAddressMonitor();
		
		m.start("/service/" + BusServer.class.getName());
		
		try {
			Thread.sleep(100000000);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
